403Webshell
Server IP : 121.121.20.254  /  Your IP : 216.73.216.202
Web Server : Microsoft-IIS/10.0
System : Windows NT WEB-SERVER 10.0 build 20348 (Windows Server 2022) AMD64
User : IUSR ( 0)
PHP Version : 8.3.28
Disable Function : NONE
MySQL : ON  |  cURL : ON  |  WGET : OFF  |  Perl : OFF  |  Python : OFF  |  Sudo : OFF  |  Pkexec : OFF
Directory :  C:/Python313/Lib/test/__pycache__/

Upload File :
current_dir [ Writeable ] document_root [ Writeable ]

 

Command :


[ Back ]     

Current File : C:/Python313/Lib/test/__pycache__/test_threading.cpython-313.opt-1.pyc
�

��i�F����SrSSKrSSKJrJrJr SSKJrJrJr SSK	J
r
 SSKJrJ
r
 SSKJr SSKrSSKrSSKrSSKrSSKrSSKrSSKrSSKrSSKrSSKrSSKrSSKrSSKrSSKJr SS	KJr SS
KJr SSKJr \RB"SS
9 Sr"Sr#Sr$Sr%"SS\&5r'"SS\RP5r)"SS\RT5r+"SS\+5r,"SS\+5r-"SS\+5r."SS\+5r/"S S!\RP5r0"S"S#\+5r1"S$S%\+5r2"S&S'\Rf5r3"S(S)\Rh5r5\Rl"\RnSLS*5"S+S,\Rh55r8"S-S.\Rr5r9"S/S0\Rh5r:"S1S2\Rv5r;"S3S4\Rx5r<"S5S6\Rz5r="S7S8\R|5r>"S9S:\RT5r?"S;S<\RT5r@"S=S>\RT5rA\BS?:Xa\R�"5 gg!\ a SrGN�f=f)@z!
Tests for the threading module.
�N)�threading_helper�requires_subprocess�requires_gil_enabled)�verbose�cpython_only�	os_helper)�
import_module)�assert_python_ok�assert_python_failure)�force_not_colorized)�mock)�
lock_tests)�support)�interpretersT)�module)�netbsd5zhp-ux11c��[R(d[R"S5"U5$[R
[;a[R"S5"U5$[R(a[R"S5"U5$[R"SS9(a[R"S5"U5$U$)Nzrequires working os.fork()z*due to known OS bug related to thread+forkz?libasan has a pthread_create() dead lock related to thread+forkT��threadz'TSAN doesn't support threads after fork)	r�has_fork_support�unittest�skip�sys�platform�platforms_to_skip�HAVE_ASAN_FORK_BUG�check_sanitizer)�tests �(C:\Python313\\Lib\test\test_threading.py�skip_unless_reliable_forkr ,s����#�#��}�}�9�:�4�@�@�
�|�|�(�(��}�}�I�J�4�P�P��!�!��}�}�^�_�`d�e�e����d�+��}�}�F�G��M�M��K�c�H�[R"[SLS5"U5$)z>Decorator to skip a test if subinterpreters are not supported.Nzsubinterpreters required)r�skipIfr)�meths r�requires_subinterpretersr%8s%���?�?�<�4�/�5�7�7;�=�=r!c��UR[[S[R5 [R[lg)N�
excepthook)�
addCleanup�setattr�	threadingr'�__excepthook__)�testcases r�restore_default_excepthookr->s*�������L�)�:N�:N�O�$�3�3�I�r!c�,�\rSrSrSrSrSrSrSrg)�Counter�Dc��SUlg�Nr��value��selfs r�__init__�Counter.__init__Es	����
r!c�.�U=RS-
slg�N�r3r5s r�inc�Counter.incG����
�
�a��
r!c�.�U=RS-slgr:r3r5s r�dec�Counter.decIr>r!c��UR$�Nr3r5s r�get�Counter.getKs���z�z�r!r3N)	�__name__�
__module__�__qualname__�__firstlineno__r7r<r@rD�__static_attributes__�r!rr/r/Ds�����r!r/c� �\rSrSrSrSrSrg)�
TestThread�Nc�n�[RRXS9 X lX0lX@lXPlg)N��name)r*�Threadr7r,�sema�mutex�nrunning)r6rQr,rSrTrUs      rr7�TestThread.__init__Os-�����!�!�$�!�2� �
��	��
� �
r!c���[R"5S-n[(a[SURUS-4-5 UR UR
 URR5 [(a$[URR5S5 URRURR5S5 SSS5 [R"U5 [(a[SURS5 UR
 URR5 URRURR5S5 [(a2[S	URURR54-5 SSS5 SSS5 g!,(df   N�=f!,(df   N(=f!,(df   g=f)
Ng��@ztask %s will run for %.1f usecg��.Aztasks are running��task�donerz$%s is finished. %d tasks are running)�randomr�printrQrSrTrUr<rDr,�assertLessEqual�time�sleepr@�assertGreaterEqual)r6�delays  r�run�TestThread.runVsE���
�
��'�)���7��2��9�9�e�c�k�*�+�
,��Y�Y�����
�
�!�!�#��7��$�-�-�+�+�-�/B�C��
�
�-�-�d�m�m�.?�.?�.A�1�E�	�
�J�J�u���w��f�d�i�i��0�����
�
�!�!�#��
�
�0�0����1B�1B�1D�a�H��7��@��9�9�d�m�m�&7�&7�&9�:�;�<�	��Y��������Y�s?�

G#�A>G�AG#�$BG�0G#�
G	�G#�
G 	�G#�#
G1)rTrUrSr,N)rFrGrHrIr7rbrJrKr!rrMrMNs��!�<r!rMc� �\rSrSrSrSrSrg)�BaseTestCase�oc�8�[R"5UlgrC)r�threading_setup�_threadsr5s r�setUp�BaseTestCase.setUpps��(�8�8�:��
r!c�z�[R"UR6 [RR5 grC)r�threading_cleanuprirr�
reap_childrenr5s r�tearDown�BaseTestCase.tearDownss$���*�*�D�M�M�:����"�"�$r!)riN)rFrGrHrIrjrorJrKr!rrereos��;�%r!rec��\rSrSrSr\S5rSrSrSr	Sr
SrS	rS
r
SrSrS
rSrSrSrSrSr\R."SSSS9S5rSrSrSrSrSr\S5r\S5r Sr!\\"RF"\$"\%S 5S!5S"55r&\\"RF"\$"\%S 5S!5S#55r'\\"RF"\$"\%S 5S!5S;S$j55r(S;S%jr)S&r*S'r+S(r,S)r-\S*5r.S+r/S,r0S-r1S.r2S/r3S0r4S1r5S2r6S3r7S4r8\\"RF"\$"\9S55S65S755r:\S85r;S9r<g:)<�ThreadTests�xi'c��Sn[R"SS9nURURS5 [R"SS9nURURS5 [R"USS9nURURS5 [R
R
[SS	S
9 [R"SS9nURURS5 SSS5 [R
R
[SS
S
9 [R"5nURURS5 SSS5 [R
R
[SSS
9 [R"US9nURURS5 SSS5 g!,(df   N�=f!,(df   N|=f!,(df   g=f)Nc��grCrKrKr!r�func�#ThreadTests.test_name.<locals>.func}���Dr!�myname1rP�{�123�myname2)�targetrQ�_counter�)�return_value�zThread-2rXzThread-3��r}zThread-5 (func))r*rR�assertEqualrQr
�patch�object)r6rvrs   r�	test_name�ThreadTests.test_name{sd����!�!�y�1��������i�0��!�!�s�+��������e�,��!�!��I�>��������i�0�
�Z�Z�
�
�y�*�1�
�
E��%�%�2�.�F����V�[�[�*�5�F��Z�Z�
�
�y�*�1�
�
E��%�%�'�F����V�[�[�*�5�F��Z�Z�
�
�y�*�1�
�
E��%�%�T�2�F����V�[�[�*;�<�F�
E�F�
E��F�
E��F�
E�s$�81F+�2F<�11G
�+
F9�<
G
�
Gc�R^�S/nSnS/nSnS/4nS/nUU4Sj4UU4Sj4UU4Sj4UU4Sj4UU4S	j4UU4S
j44nUHQup�TRX�S9 [R"X�S9n
U
R5 U
R	5 SSS5 MS g!,(df   Me=f)Nr;�r;�str�r�c�(>�TRUS5$r:�r���argr6s �r�<lambda>�0ThreadTests.test_args_argument.<locals>.<lambda>�s���4�#3�#3�C��#;r!c�(>�TRUS5$r:r�r�s �rr�r��s���D�$4�$4�S�!�$<r!c�(>�TRUS5$�Nr�r�r�s �rr�r��s���4�#3�#3�C��#?r!c�(>�TRUS5$r�r�r�s �rr�r��s���D�$4�$4�S�%�$@r!c�*>�TRUS/5$r:r�r�s �rr�r��s����(8�(8��q�c�(Br!c�(>�TRUS5$)Nr�r�r�s �rr�r��s����(8�(8��d�(Cr!�r}�args)�subTestr*rR�start�join)r6�num_list�	num_tuple�str_list�	str_tuple�
list_in_tuple�
tuple_in_list�
test_casesr�r}�ts`          r�test_args_argument�ThreadTests.test_args_argument�s�����3���	��7���	����
���
��;�<�
�<�=�
�?�@�
�@�A�
�B�C�
�C�D�

�
�'�L�D����V��7��$�$�F�>�����	�����8�7�'�7�7�s�5B�
B&	c	�
�[R"5 UR[[RS5 UR[[RSS9 UR[[RSSSSS9 g)Nr;)�ar)r��b)r*�Lock�assertRaises�	TypeErrorr5s r�test_lock_no_args�ThreadTests.test_lock_no_args�sZ���������)�Y�^�^�Q�7����)�Y�^�^�q��9����)�Y�^�^�Q��Q�!��Dr!c��UR[5 "SS[R5nSSS5 g!,(df   g=f)Nc��\rSrSrSrg)�1ThreadTests.test_lock_no_subclass.<locals>.MyLock�rKN�rFrGrHrIrJrKr!r�MyLockr��s��$r!r�)r�r�r*r�)r6r�s  r�test_lock_no_subclass�!ThreadTests.test_lock_no_subclass�s*���
�
�y�
)�.����.�*�
)�
)�s	�9�
Ac�f�SSKnUR[RS-UR5 gr2)�types�assertIsInstancer*r��	UnionType)r6r�s  r�test_lock_or_none�ThreadTests.test_lock_or_none�s"������i�n�n�t�3�U�_�_�Er!c���Sn[R"SS9n[R"5n[5n/n[	U5Hjn[SU-XX45nUR
U5 URUR5 UR[U5S5 UR5 Ml [[S5(aY[SU55[R"51-nURSU5 UR![#U5US-5 [$(a['S	5 UH�nUR)5 UR+UR-55 UR/URS
5 UR1UR5 UR[U5S5 M� [$(a['S5 UR!UR35S
5 g)
N�
rXr3z<thread %d>z^<TestThread\(.*, initial\)>$�
get_native_idc3�8# �UHoRv� M g7frC)�	native_id)�.0r�s  r�	<genexpr>�/ThreadTests.test_various_ops.<locals>.<genexpr>�s���:�'�Q�[�[�'�s�r;z!waiting for all tasks to completerz#^<TestThread\(.*, stopped -?\d+\)>$zall tasks done)r*�BoundedSemaphore�RLockr/�rangerM�append�assertIsNone�ident�assertRegex�reprr��hasattr�setr��assertNotInr��lenrr\r��assertFalse�is_alive�assertNotEqual�assertIsNotNonerD)	r6�NUMTASKSrSrT�
numrunning�threads�ir��
native_idss	         r�test_various_ops�ThreadTests.test_various_ops�s~�����)�)��2�����!���Y�
����x��A��=��?�D��J�A��N�N�1�����a�g�g�&����T�!�W�&F�G�
�G�G�I�!��9�o�.�.��:�'�:�:�i�>U�>U�>W�=X�X�J����T�:�.����S��_�h��l�;��7��5�6��A�
�F�F�H����Q�Z�Z�\�*��������+�� � ����)����T�!�W�&L�M���7��"�#�������)�1�-r!c�v^^�UR[R"5R5 UU4Sjn[R"5m/m[
R"5 [R"US5nTR5 URTSU5 SSS5 g!,(df   g=f)Nc�>�TR[R"5R5 TR	5 grC)r�r*�current_threadr�r�)rZr�s��r�f�9ThreadTests.test_ident_of_no_threading_threads.<locals>.f�s'����L�L��1�1�3�9�9�:��H�H�Jr!rKr)r�r*r�r��Eventr�wait_threads_exit�_thread�start_new_thread�waitr�)r6r��tidrZr�s   @@r�"test_ident_of_no_threading_threads�.ThreadTests.test_ident_of_no_threading_threads�s������Y�5�5�7�=�=�>�	���� ����
�
/�
/�
1��*�*�1�b�1�C��I�I�K����U�1�X�s�+�2�
1�
1�s�$=B*�*
B8c��[(a[S5 [R"S5 UR5 [R"S5 g![R
a [R"S5ef=f)Nz!with 256 KiB thread stack size...i�4platform does not support changing thread stack sizer�	rr\r*�
stack_sizer��errorr�SkipTestr�r5s r�test_various_ops_small_stack�(ThreadTests.test_various_ops_small_stack�so���7��5�6�	H�� � ��(�	
�������Q���	�}�}�	H��#�#�F�H�
H�	H���A�+Bc��[(a[S5 [R"S5 UR5 [R"S5 g![R
a [R"S5ef=f)Nzwith 1 MiB thread stack size...ir�rr�r5s r�test_various_ops_large_stack�(ThreadTests.test_various_ops_large_stackso���7��3�4�	H�� � ��*�	
�������Q���	�}�}�	H��#�#�F�H�
H�	H�r�c��^^^�SmSmUUU4Sjn[R"5nUR5 [R"5 [
R"X45nUR5 SSS5 TbTeTRWTR5 TR[5 TR5 SSS5 TRR5 TR[5 TR5 SSS5 Sn[R "5U-n[R "5U:a^[R"R%TR5TLag[R&"S5 [R "5U:aM^TR)S5 g!,(df   GNI=f!,(df   GN=f!,(df   N�=f)Nc�>�[R"5mTRnTRU[R5 TRT[R5 TR[RRU5T5 TRTR5S5 TRS[T55 UR5 g![anUmSnAN"SnAff=f!UR5 f=f)Nz-Expected _DummyThread to be considered alive.�_DummyThread)r*r�r��assertIn�_activer�r��assertIsrD�
assertTruer�r��
BaseException�release)rTr��e�dummy_threadr�r6s   ���rr��*ThreadTests.test_foreign_thread.<locals>.fs����
 �
 )�7�7�9��"�(�(���
�
�c�9�#4�#4�5��%�%�l�I�4J�4J�K��
�
�i�/�/�3�3�C�8�,�G���� �)�)�+�C���
�
�n�d�<�.@�A��
�
���!�
����
���
�
��s*�CC�
C5�)C0�+C8�0C5�5C8�8D
r�皙�����?z[It was expected that the created threading._DummyThread was removed from threading._active.)r*r��acquirerr�r�r�r�r�r��RuntimeErrorr��_started�clearr�r^�	monotonicr�rDr_�fail)r6r�rTr��timeout�
timeout_atrr�s`     @@r�test_foreign_thread�ThreadTests.test_foreign_threadsW�������	 �,��� ��
�
�
��
�
/�
/�
1��*�*�1�h�7�C��M�M�O�2����K�����l�0�0�1�
�
�
�|�
,�����-����#�#�%�
�
�
�|�
,��!�!�#�-����^�^�%��/�
��n�n���+�� � �$�$�\�%7�%7�8��L���J�J�r�N��n�n���+�

�I�I�s�t�-2�
1��-�
,��-�
,�s$�
(F4�0G�8G�4
G�
G�
G&c�f^	^
^�[S5nURRnURUR4Ul"SS[5m	UR	T	5n[R"5nURU[5 URUS5 U"XC5nM!T	a Of=fURWS5 O![a Of=f[R"5m[R"5m
"U	U
U4SjS[R 5nU"5nSUlUR%5 [&(a[)S	5 [&(a[)S
5 U"SU5nURUS5 [&(a[)S5 TR+5nUR-U5 [&(a[)S
5 UR/UR05 [&(a[)S5 U"UR2U5nURUS5 [&(a[)S5 T
R+[4R6S9 UR-UR05 [&(a[)S5 UR0(aUR95 gg)N�ctypesc��\rSrSrSrg)�<ThreadTests.test_PyThreadState_SetAsyncExc.<locals>.AsyncExciHrKNr�rKr!r�AsyncExcrHs��r!rrTr;c�&>�\rSrSrUUU4SjrSrg)�:ThreadTests.test_PyThreadState_SetAsyncExc.<locals>.Workerikc��>�[R"5UlSUlTR	5 [
R"S5 M(!Ta SUlTR	5 gf=f)NFTr)r*�	get_ident�id�finishedr�r^r_)r6r�worker_saw_exception�worker_starteds ���rrb�>ThreadTests.test_PyThreadState_SetAsyncExc.<locals>.Worker.runlsb���#�-�-�/��� %��
�/��&�*�*�,��
�
�3���� �/�$(�D�M�(�,�,�.�/�s�)A
�
A-�,A-)rrN�rFrGrHrIrbrJ)rrrs���r�Workerrks
���

/�

/r!rz    started worker threadz     trying nonsensical thread id���z,    waiting for worker thread to get startedz"    verifying worker hasn't exitedz2    attempting to raise asynch exception in workerz5    waiting for worker to say it caught the exception�r	z    all OK -- joining worker)r	�	pythonapi�PyThreadState_SetAsyncExc�c_ulong�	py_object�argtypes�	Exceptionr*rr��int�
assertGreaterrr��UnboundLocalErrorr�rR�daemonr�rr\r�r�r�rrr�
SHORT_TIMEOUTr�)r6r�
set_async_exc�	exceptionr��resultrr��retrrrs         @@@r�test_PyThreadState_SetAsyncExc�*ThreadTests.test_PyThreadState_SetAsyncExcBs-����x�(���(�(�B�B�
�"(�.�.�&�2B�2B�!C�
��	�y�	��$�$�X�.�	��!�!�#�����c�3�'����3��"�	-�"�3�2�F������	��	��	����V�Q�'�� �	��	��#���*��(���0��	/�	/�Y�%�%�	/�
�H�����	���	��7��-�.��7��4�5��r�9�-�������#��7��@�A��!�!�#��������7��6�7�������$��7��F�G��q�t�t�Y�/�������#��7��I�J��!�!�'�*?�*?�!�@�����
�
�#��7��0�1��:�:�
�F�F�H�s$� B,�,B5�4B5�9C�
C�Cc�<�Sn[RnU[l[R"SS9nUR[RUR
5 UR
U[R;S5 U[lg!U[lf=f)Nc�,�[R"5erC)r*�ThreadError�r��kwargss  r�fail_new_thread�7ThreadTests.test_limbo_cleanup.<locals>.fail_new_thread�s���'�'�)�)r!c��grCrKrKr!rr��0ThreadTests.test_limbo_cleanup.<locals>.<lambda>����r!r�z:Failed to cleanup _limbo map on failure of Thread.start().)r*�_start_joinable_threadrRr�r2r�r��_limbo)r6r5r:r�s    r�test_limbo_cleanup�ThreadTests.test_limbo_cleanup�s���	*�!*�!A�!A��+:�	�(�	F�� � ��5�A����i�3�3�Q�W�W�=�����Y�%�%�%�L�
N�0F�I�,��/E�I�,�s�A"B�
Bc��[R"SS9(aURS5 [S5 [	SS5upnURUS5 g)NTr�TSAN would report thread leakr�-caNif 1:
            import ctypes, sys, time, _thread

            # This lock is used as a simple event variable.
            ready = _thread.allocate_lock()
            ready.acquire()

            # Module globals are cleared before __del__ is run
            # So we save the functions in class dict
            class C:
                ensure = ctypes.pythonapi.PyGILState_Ensure
                release = ctypes.pythonapi.PyGILState_Release
                def __del__(self):
                    state = self.ensure()
                    self.release(state)

            def waitingThread():
                x = C()
                ready.release()
                time.sleep(100)

            _thread.start_new_thread(waitingThread, ())
            ready.acquire()  # Be sure the other thread is waiting.
            sys.exit(42)
            �*)rr�skipTestr	rr��r6�rc�out�errs    r�test_finalize_running_thread�(ThreadTests.test_finalize_running_thread�sP���"�"�$�/�
�M�M�9�:��h��,�T�4�����2	
����R� r!c�p�[R"SS9(aURS5 [SS5 g)NTrr?r@aPif 1:
            import sys, threading

            # A deadlock-killer, to prevent the
            # testsuite to hang forever
            def killer():
                import os, time
                time.sleep(2)
                print('program blocked; aborting')
                os._exit(2)
            t = threading.Thread(target=killer)
            t.daemon = True
            t.start()

            # This is the trace function
            def func(frame, event, arg):
                threading.current_thread()
                return func

            sys.settrace(func)
            )rrrBr
r5s r�test_finalize_with_trace�$ThreadTests.test_finalize_with_trace�s0���"�"�$�/�
�M�M�9�:��� �	r!c��[SS5upnURUR5S5 URUS5 g)Nr@a�if 1:
                import threading
                from time import sleep

                def child():
                    sleep(1)
                    # As a non-daemon thread we SHOULD wake up and nothing
                    # should be torn down yet
                    print("Woke up, sleep function is:", sleep)

                threading.Thread(target=child).start()
                raise SystemExit
            s5Woke up, sleep function is: <built-in function sleep>r!�r
r��striprCs    r�test_join_nondaemon_on_shutdown�+ThreadTests.test_join_nondaemon_on_shutdown�sE��(��/�����	
�������D�	F�����c�"r!c��[Rn[R"5n[	SS5Hnn[
R"US-5 [R"SS9nUR5 UR5 U"5nURXESX54-5 Mp [R"U5 g![R"U5 f=f)Nr;�dg-C��6*?c��grCrKrKr!rr��7ThreadTests.test_enumerate_after_join.<locals>.<lambda>s��Dr!r�z&#1703448 triggered after %d trials: %s)r*�	enumerater�getswitchintervalr�r�setswitchintervalrRr�r�r�)r6�enum�old_intervalr�r��ls      r�test_enumerate_after_join�%ThreadTests.test_enumerate_after_joins����"�"���,�,�.��
	0��1�c�]���)�)�!�f�*�5��$�$�L�9�����	������F��� � ��<��v�E�G�
#�
�!�!�,�/��C�!�!�,�/�s�A>B<�<C�i�F��size�memuse�dry_runc	�^�/mSnU4Sjn[SS5H�n[R"US9/n[U5H+nUR[R"UUS4S95 M- UHnUR	5 M [
R"S5 UHnUR5 M T(dM�TSe g)	Nc�0�[R"S5 g)Ng{�G�zt?�r^r_rKr!r�worker�;ThreadTests.test_join_from_multiple_threads.<locals>.workers���J�J�u�r!c�x>�UR5 g![anTRU5 SnAgSnAff=frC)r�r$r�)rr��errorss  �r�joiner�;ThreadTests.test_join_from_multiple_threads.<locals>.joiners0���
!����
���
!��
�
�a� � ��
!�s��
9�4�9rr]r�rr��{�G�z�?)r�r*rRr�r�r^r_r�)	r6r_reri�Nr�r�r�rhs	        @r�test_join_from_multiple_threads�+ThreadTests.test_join_from_multiple_threadss������	�	!��q�"��A� �'�'�v�6�7�G��1�X�����y�/�/�v�6=�a�j�]� D�E�������	���J�J�t����������v��Q�i��r!c�^�[R"5mTR5 U4Sjn[R"US9nUR5 UR
SS9 TR5 UR
5 g)Nc�&>�TR5 grC)r)�locks�rre�2ThreadTests.test_join_with_timeout.<locals>.worker0s����L�L�Nr!r�rkr)r��
allocate_lockrr*rRr�r�r�)r6rerrqs   @r�test_join_with_timeout�"ThreadTests.test_join_with_timeout,sZ����$�$�&������	��!�!��0���������D��!��������
r!c���"SS[5n[U5 U"SS9n[R"U5nURR5 AUR
U"5S[R"U"55-S9 U"SS9n[R"U5nURR5 AUR
U"5S[R"U"55-S9 g)Nc� �\rSrSrSrSrSrg)�DThreadTests.test_no_refcycle_through_target.<locals>.RunSelfFunctioni<c��Xl[R"URU4SU0S9UlURR5 g)N�yet_another)r}r�r4)�should_raiser*rR�_runrr�)r6r{s  rr7�MThreadTests.test_no_refcycle_through_target.<locals>.RunSelfFunction.__init__=sB��%1�!�'�.�.�d�i�i�59�G�7D�T�6J�L������!�!�#r!c�2�UR(a[egrC)r{�
SystemExit)r6�	other_refrzs   rr|�IThreadTests.test_no_refcycle_through_target.<locals>.RunSelfFunction._runFs���$�$�$�$�%r!)r{rN)rFrGrHrIr7r|rJrKr!r�RunSelfFunctionrx<s��
$�
%r!r�F)r{z%d references still around)�msgT)	r�r-�weakref�refrr�r�r�getrefcount)r6r��
cyclic_object�weak_cyclic_object�raising_cyclic_object�weak_raising_cyclic_objects      r�test_no_refcycle_through_target�+ThreadTests.test_no_refcycle_through_target;s���	%�f�	%�	#�4�(�'�U�;�
�$�[�[��7�����!�!�#�����,�.�:�!�o�o�.@�.B�C�D�	�	F�!0�T� B��%,�[�[�1F�%G�"��$�$�)�)�+�!����4�6�:�!�o�o�.H�.J�K�L�	�	Nr!c���[R"5nUR[S5 UR	5 SSS5 UR[S5 URS5 SSS5 UR[S5 UR
5 SSS5 UR[S5 URS5 SSS5 [R"5nUR[S5 UR5 SSS5 [R"5nUR5 UR[S5 UR5 SSS5 UR[S	5 [R"5 SSS5 UR[S
5 [R"5 SSS5 g!,(df   GN�=f!,(df   GN�=f!,(df   GNd=f!,(df   GNF=f!,(df   GN=f!,(df   N�=f!,(df   N�=f!,(df   g=f)Nzget the daemon attributezset the daemon attributeTzget the name attributezset the name attributerQzuse is_set()zuse notify_all()zuse active_count()zuse current_thread())r*rR�assertWarnsRegex�DeprecationWarning�isDaemon�	setDaemon�getName�setNamer��isSet�	Conditionr�	notifyAll�activeCount�
currentThread)r6r�r��conds    r�test_old_threading_api�"ThreadTests.test_old_threading_api\s���
�����
�
"�
"�#5�#>�@�
�J�J�L�@��
"�
"�#5�#>�@�
�K�K���@��
"�
"�#5�#<�>�
�I�I�K�>��
"�
"�#5�#<�>�
�I�I�f��>�
�O�O���
�
"�
"�#5�~�
F�
�G�G�I�G��"�"�$������
�
"�
"�#5�7I�
J��N�N��K��
"�
"�#5�7K�
L��!�!�#�M�
�
"�
"�#5�7M�
N��#�#�%�O�
N�1@�@��@�@��>�>��>�>��
G�
F��
K�
J��M�
L��
N�
N�s_�G�G'�G9�:H�?H�H/�I�6I�
G$�'
G6�9
H�
H�
H,�/
H=�
I�
Ic��[R"5nURS[U55 SUlURS[U55 g�Nr(T)r*rRr�r�r(r��r6r�s  r�test_repr_daemon�ThreadTests.test_repr_daemon{s=�����������4��7�+�����
�
�h��Q��(r!c� �[R"5nURUR5 [R"SS9nURUR5 [R"SS9nUR	UR5 g)NF)r(T)r*rRr�r(r�r�s  r�test_daemon_param�ThreadTests.test_daemon_param�sd�������������"����E�*��������"����D�)��������!r!c�n�Sn[SU5up#nURUS5 URUS5 g)Na�if 1:
            import _thread, threading, os, time, warnings

            def background_thread(evt):
                # Creates and registers the _DummyThread instance
                threading.current_thread()
                evt.set()
                time.sleep(10)

            evt = threading.Event()
            _thread.start_new_thread(background_thread, (evt,))
            evt.wait()
            assert threading.active_count() == 2, threading.active_count()
            with warnings.catch_warnings(record=True) as ws:
                warnings.filterwarnings(
                        "always", category=DeprecationWarning)
                if os.fork() == 0:
                    assert threading.active_count() == 1, threading.active_count()
                    os._exit(0)
                else:
                    assert ws[0].category == DeprecationWarning, ws[0]
                    assert 'fork' in str(ws[0].message), ws[0]
                    os.wait()
        r@r!�r
r��r6�code�_rErFs     r�test_dummy_thread_after_fork�(ThreadTests.test_dummy_thread_after_fork�s;����0'�t�T�2��������c�"�����c�"r!c�^�[R"5nUR[RU5 [R
RS5 [
S5H�n[R"SS9nUR5 [R"[SS9 [R"5=nS:Xa.[R"UR!5(aSOS	5 O%UR#5 [
R$"US	S
9 SSS5 M� g!,(df   M�=f)Ng���ư>r]c��grCrKrKr!rr��6ThreadTests.test_is_alive_after_fork.<locals>.<lambda>�r9r!r��ignore��category�actionr�r���exitcode)rrVr(rWrrr�r*rRr��warnings�catch_warningsr��os�fork�_exitr�r��wait_process)r6rYr�r��pids     r�test_is_alive_after_fork�$ThreadTests.test_is_alive_after_fork�s����,�,�.������-�-�|�<�	
���&�&�t�,��r��A�� � ��5�A�
�G�G�I��(�(�2D�08�:��7�7�9�$�C��*��H�H�1�:�:�<�<�R�R�8��F�F�H��(�(��r�:�:�:�	�:�:�s
�!A/D�
D,	c�^�[R"5nTRURS5 TRUR[R
"5R5 TRUR[R"55 U4Sjn[R"US9nUR5 UR5 g)N�
MainThreadc�>�TR[R"5R[R"5R5 grC)r�r*�main_threadr�r�r5s�rr��'ThreadTests.test_main_thread.<locals>.f�s5������	� 5� 5� 7� =� =� )� 8� 8� :� @� @�
Br!r�)
r*r�r�rQr�r�rrRr�r�)r6�mainr��ths`   r�test_main_thread�ThreadTests.test_main_thread�s�����$�$�&��������L�1�������Y�%=�%=�%?�%E�%E�F�������Y�%8�%8�%:�;�	B��
�
�Q�
'��
���
�
���	r!�waitpidztest needs os.waitpid()c��Sn[SU5up#nUR5RSS5nURUS5 URUS5 g)Na!if 1:
            import os, threading
            from test import support

            ident = threading.get_ident()
            pid = os.fork()
            if pid == 0:
                print("current ident", threading.get_ident() == ident)
                main = threading.main_thread()
                print("main", main.name)
                print("main ident", main.ident == ident)
                print("current is main", threading.current_thread() is main)
            else:
                support.wait_process(pid, exitcode=0)
        r@�
r�r!zHcurrent ident True
main MainThread
main ident True
current is main True
�r
�decode�replacer��r6r�r�rErF�datas      r�test_main_thread_after_fork�'ThreadTests.test_main_thread_after_fork�sX����'�t�T�2�����z�z�|�#�#�D�"�-������c�"�����2�	3r!c���Sn[SU5up#nUR5RSS5nURURS5S5 URUS5 g)Na�if 1:
            import os, threading, sys, warnings
            from test import support

            def func():
                ident = threading.get_ident()
                with warnings.catch_warnings(record=True) as ws:
                    warnings.filterwarnings(
                            "always", category=DeprecationWarning)
                    pid = os.fork()
                    if pid == 0:
                        print("current ident", threading.get_ident() == ident)
                        main = threading.main_thread()
                        print("main", main.name, type(main).__name__)
                        print("main ident", main.ident == ident)
                        print("current is main", threading.current_thread() is main)
                        # stdout is fully buffered because not a tty,
                        # we have to flush before exit.
                        sys.stdout.flush()
                    else:
                        assert ws[0].category == DeprecationWarning, ws[0]
                        assert 'fork' in str(ws[0].message), ws[0]
                        support.wait_process(pid, exitcode=0)

            th = threading.Thread(target=func)
            th.start()
            th.join()
        r@r�r��utf-8zTcurrent ident True
main Thread-1 (func) Thread
main ident True
current is main True
r�r�s      r�/test_main_thread_after_fork_from_nonmain_thread�;ThreadTests.test_main_thread_after_fork_from_nonmain_thread�sa����8'�t�T�2�����z�z�|�#�#�D�"�-��������G�,�b�1�����2�	r!c��SU-n[SSSU5up4nUR5RSS5nURUR5S5 URUU(aSOSSU<S	3-S
-5 g)NaCif 1:
            import os, threading, sys, traceback, _thread
            from test import support

            def func(lock):
                ident = threading.get_ident()
                if %s:
                    # call current_thread() before fork to allocate DummyThread
                    current = threading.current_thread()
                    print("current", current.name, type(current).__name__)
                print("ident in _active", ident in threading._active)
                # flush before fork, so child won't flush it again
                sys.stdout.flush()
                pid = os.fork()
                if pid == 0:
                    print("current ident", threading.get_ident() == ident)
                    main = threading.main_thread()
                    print("main", main.name, type(main).__name__)
                    print("main ident", main.ident == ident)
                    print("current is main", threading.current_thread() is main)
                    print("_dangling", [t.name for t in list(threading._dangling)])
                    # stdout is fully buffered because not a tty,
                    # we have to flush before exit.
                    sys.stdout.flush()
                    try:
                        threading._shutdown()
                        os._exit(0)
                    except:
                        traceback.print_exc()
                        sys.stderr.flush()
                        os._exit(1)
                else:
                    try:
                        support.wait_process(pid, exitcode=0)
                    except Exception:
                        # avoid 'could not acquire lock for
                        # <_io.BufferedWriter name='<stderr>'> at interpreter shutdown,'
                        traceback.print_exc()
                        sys.stderr.flush()
                    finally:
                        lock.release()

            join_lock = _thread.allocate_lock()
            join_lock.acquire()
            th = _thread.start_new_thread(func, (join_lock,))
            join_lock.acquire()
        z-Wzignore::DeprecationWarningr@r�r�zcurrent Dummy-1 _DummyThread
zident in _active �
zmcurrent ident True
main MainThread _MainThread
main ident True
current is main True
_dangling ['MainThread']
r�)r6�create_dummyr�r�rErFr�s       r�/test_main_thread_after_fork_from_foreign_thread�;ThreadTests.test_main_thread_after_fork_from_foreign_threads���.�\�].��b'�t�-I�4�QU�V�����z�z�|�#�#�D�"�-���������r�*�����>J�:�PR�,�\�,<�B�?�@�6�6�	7r!c�"�URSS9 g)NT)r�)r�)r6r�s  r�-test_main_thread_after_fork_from_dummy_thread�9ThreadTests.test_main_thread_after_fork_from_dummy_threadQs���<�<�$�<�Or!c��Sn[SU5up#nUR5nURUS5 URUR5S/S-5 g)Na�if 1:
            import gc, threading

            main_thread = threading.current_thread()
            assert main_thread is threading.main_thread()  # sanity check

            class RefCycle:
                def __init__(self):
                    self.cycle = self

                def __del__(self):
                    print("GC:",
                          threading.current_thread() is main_thread,
                          threading.main_thread() is main_thread,
                          threading.enumerate() == [main_thread])

            RefCycle()
            gc.collect()  # sanity check
            x = RefCycle()
        r@r!zGC: True True Truer)r
r�r��
splitlinesr�s      r� test_main_thread_during_shutdown�,ThreadTests.test_main_thread_during_shutdownTsX����('�t�T�2�����z�z�|������c�"�������*�.�/�!�3�	5r!c�J�Sn[SU5up#nURUS5 g)Na�if 1:
            import os
            import threading
            import time
            import random

            def random_sleep():
                seconds = random.random() * 0.010
                time.sleep(seconds)

            class Sleeper:
                def __del__(self):
                    random_sleep()

            tls = threading.local()

            def f():
                # Sleep a bit so that the thread is still running when
                # Py_Finalize() is called.
                random_sleep()
                tls.x = Sleeper()
                random_sleep()

            threading.Thread(target=f).start()
            random_sleep()
        r@r!r��r6r�rDrErFs     r�test_finalization_shutdown�&ThreadTests.test_finalization_shutdownqs+����4(��d�3��������c�"r!c�4^^�[R"5m[R"5mTR5 TR5 UU4Sjn[R"US9nUR5 TR5 UR
S[U55 TR5 Sn[S5H*nU[U5;a O[R"S5 M, UR
U[U55 UR5 g)Nc�F>�TR5 TR5 grC)r�r)�finish�starteds��rr��(ThreadTests.test_repr_stopped.<locals>.f�s����O�O���N�N�r!r�r��stoppedi�rk)
r�rsrr*rRr�r�r�r�r�r^r_r�)r6r�r��LOOKING_FORr�r�r�s     @@r�test_repr_stopped�ThreadTests.test_repr_stopped�s�����'�'�)���&�&�(����������	�
���A�&��	���	������
�
�i��a��)����� ���s��A��d�1�g�%���J�J�t���	
�
�
�k�4��7�+�	���r!c�R�[SS5GHn[R"U5n[U5Vs/sH!n[R"URS9PM# nnUHnUR5 M UHnUR
5 M [U5Vs/sH!n[R"URS9PM# nnUHnUR5 M UHnUR
5 M UR[UR5 GM gs snfs snf)Nr;r�r�)
r�r*r�rRrr�r�r�r��
ValueError)r6�limit�bsr�r�r�s      r�test_BoundedSemaphore_limit�'ThreadTests.test_BoundedSemaphore_limit�s����1�b�\�E��+�+�E�2�B� %�e��.� ,�1�!�'�'�r�z�z�:� ,�
�.������	���������!&�e��.� ,�1�!�'�'�r�z�z�:� ,�
�.������	������������j�"�*�*�5�"��.��.s�(D�(D$c��^^^�[S5nU4SjmSmUU4SjmSTl[R"5n[R"T5 [
R"T5 UR
T5 [S5H
nT"5 M [R"U5 [
R"U5 g![R"U5 [
R"U5 f=f)N�	_testcapic�>�T$rCrK��frame�eventr��
noop_traces   �rr��9ThreadTests.test_frame_tstate_tracing.<locals>.noop_trace��	����r!c3�# �Sv� M7f)N�	generatorrKrKr!rr��8ThreadTests.test_frame_tstate_tracing.<locals>.generator�s����!�!��s�
c�`>�TRcT"5Tl[TR5$rC)�gen�next)�callbackr�s��rr�7ThreadTests.test_frame_tstate_tracing.<locals>.callback�s%����|�|�#�(�{�������%�%r!rX)r	rr�gettrace�settracer*�call_in_temporary_c_threadr�)r6r��	old_tracerrr�r�s    @@@r�test_frame_tstate_tracing�%ThreadTests.test_frame_tstate_tracing�s����!�+�.�	�	�	"�	&�����L�L�N�	����Z� �	*����z�*�
�0�0��:��a����
�!�
�L�L��#����y�)��
�L�L��#����y�)�s
�AB?�?.C-c�^�U4Sjm[R"5n[R"T5 [R"5nURTU5 [R"U5 g![R"U5 f=f)Nc�>�T$rCrKr�s   �rr��-ThreadTests.test_gettrace.<locals>.noop_trace�r�r!)r*rrr�)r6r�
trace_funcr�s   @r�
test_gettrace�ThreadTests.test_gettrace�sf���	��&�&�(�	�	*����z�*�"�+�+�-�J����Z�
�3����y�)��I���y�)�s�=A2�2B
c�*^^^�Sn[R"5n[R"5m[R"5m/mUUU4Sjn[R"US9nUR	5 TR5 [R"U5 TR5 UR5 URTSU/5 UR[R"5U5 UR[R"5U5 [R"U5 UR[R"5U5 UR[R"5U5 g![R"U5 f=f)Nc��grCrK�r�s r�fn�1ThreadTests.test_gettrace_all_threads.<locals>.fn����tr!c��>�TR[R"55 TR5 TR	5 TR[R"55 grC)r�rrr�r�)�first_check�second_check�trace_funcss���r�checker�6ThreadTests.test_gettrace_all_threads.<locals>.checker�sA������s�|�|�~�.��O�O���������s�|�|�~�.r!r�)r*rr�rRr�r��settrace_all_threadsr�r�r�r)r6rrrr�rrrs     @@@r�test_gettrace_all_threads�%ThreadTests.test_gettrace_all_threads�s�����&�&�(�	��o�o�'�� ���(����	/�	6�� � ��0�A�
�G�G�I������*�*�2�.�����
�F�F�H����[�4��*�5����Y�/�/�1�2�6����S�\�\�^�R�0��*�*�9�5�����+�+�-�y�9���������3��
�*�*�9�5��
�CE:�:Fc��Sn[R"5n[R"U5 URU[R"55 [R"U5 g![R"U5 f=f)Nc��grCrKrs rr�'ThreadTests.test_getprofile.<locals>.fnrr!)r*�
getprofile�
setprofiler�)r6r�old_profiles   r�test_getprofile�ThreadTests.test_getprofiles\����*�*�,��	.�� � ��$����R��!5�!5�!7�8�� � ��-��I� � ��-�s�;A,�,Bc�*^^^�Sn[R"5n[R"5m[R"5m/mUUU4Sjn[R"US9nUR	5 TR5 [R"U5 TR5 UR5 URTSU/5 UR[R"5U5 UR[R"5U5 [R"U5 UR[R"5U5 UR[R"5U5 g![R"U5 f=f)Nc��grCrKrs rr�3ThreadTests.test_getprofile_all_threads.<locals>.fnrr!c��>�TR[R"55 TR5 TR	5 TR[R"55 grC)r�rr$r�r�)r�
profile_funcsrs���rr�8ThreadTests.test_getprofile_all_threads.<locals>.checker#sC���� � ����!1�2��O�O������� � ����!1�2r!r�)r*r$r�rRr�r��setprofile_all_threadsr�r�r�r)r6rr&rr�rr-rs     @@@r�test_getprofile_all_threads�'ThreadTests.test_getprofile_all_threadss�����*�*�,���o�o�'�� ���(���
�	3�	:�� � ��0�A�
�G�G�I������,�,�R�0�����
�F�F�H����]�T�2�J�7����Y�1�1�3�R�8����S�^�^�-�r�2��,�,�[�9�����-�-�/��=�������)�;�7��
�,�,�[�9�r c�b�[SS5upnURUR5S5 g)Nr@a(if 1:
            import threading

            class Atexit:
                def __del__(self):
                    print("thread_dict.atexit = %r" % thread_dict.atexit)

            thread_dict = threading.local()
            thread_dict.atexit = "value"

            atexit = Atexit()
        sthread_dict.atexit = 'value')r
r��rstriprCs    r�test_locals_at_exit�ThreadTests.test_locals_at_exit9s2��(��/�
����	
�������'F�Gr!c���"SS[5nU"5n[R"US9nUR5 UR	5 URUR5 g)Nc�&�\rSrSrSrSrSrSrg)�6ThreadTests.test_boolean_target.<locals>.BooleanTargetiOc��SUlg�NF��ranr5s rr7�?ThreadTests.test_boolean_target.<locals>.BooleanTarget.__init__Ps	�� ��r!c��gr:rKr5s r�__bool__�?ThreadTests.test_boolean_target.<locals>.BooleanTarget.__bool__Rs��r!c��SUlg)NTr;r5s r�__call__�?ThreadTests.test_boolean_target.<locals>.BooleanTarget.__call__Ts	����r!r;N)rFrGrHrIr7r?rBrJrKr!r�
BooleanTargetr8Os��
!�
�
 r!rDr�)r�r*rRr�r�r�r<)r6rDr}rs    r�test_boolean_target�ThreadTests.test_boolean_targetJsI��
	 �F�	 �����!�!��0���������
�����
�
�#r!c��Sn[R"5 [R"US9R	5 SSS5 g!,(df   g=f)Nc��grCrKrKr!r�noop�0ThreadTests.test_leak_without_join.<locals>.noop`rxr!r�)rr�r*rRr�)r6rIs  r�test_leak_without_join�"ThreadTests.test_leak_without_join]s6��	�
�
/�
/�
1����D�)�/�/�1�2�
1�
1�s�#A�
Ac��[R"S5n[SU5up#nURUS5 URUS5 g)Na�
            import _thread
            import sys

            event = _thread.allocate_lock()
            event.acquire()

            def import_threading():
                import threading
                event.release()

            if 'threading' in sys.modules:
                raise Exception('threading is already imported')

            _thread.start_new_thread(import_threading, ())

            # wait until the threading module is imported
            event.acquire()
            event.release()

            if 'threading' not in sys.modules:
                raise Exception('threading is not imported')

            # don't wait until the thread completes
        r@r!)�textwrap�dedentr
r�r�s     r�test_import_from_another_thread�+ThreadTests.test_import_from_another_threadesI����� �
��2(��d�3��������c�"�����c�"r!c��Sn[SU5up#nURUR5S5 URSU5 g)Na6if 1:
            import _thread

            def f():
                print("shouldn't be printed")

            class AtFinalization:
                def __del__(self):
                    print("OK")
                    _thread.start_new_thread(f, ())
            at_finalization = AtFinalization()
        r@�OKs/can't create new thread at interpreter shutdown)r
r�rNr�r�s     r�%test_start_new_thread_at_finalization�1ThreadTests.test_start_new_thread_at_finalization�s@����'�t�T�2�����������e�,��
�
�H�#�Nr!c� �SSKJnJn Sn[SSU5upVnUR
5nSU;aUR	S5 URUS5 URUS	5 g![anUR	U5 SnAN~SnAff=f)
Nr)�	setrlimit�RLIMIT_NPROCaif 1:
            import resource
            import _thread

            def f():
                print("shouldn't be printed")

            limits = resource.getrlimit(resource.RLIMIT_NPROC)
            [_, hard] = limits
            resource.setrlimit(resource.RLIMIT_NPROC, (0, hard))

            try:
                handle = _thread.start_joinable_thread(f)
            except RuntimeError:
                print('ok')
            else:
                print('!skip!')
                handle.join()
        z-ur@s!skip!z.RLIMIT_NPROC had no effect; probably superusersokr!)�resourcerWrX�ImportErrorrBr
rNr�)r6rWrXrFr�r�rEs       r�test_start_new_thread_failed�(ThreadTests.test_start_new_thread_failed�s���
	�8���&'�t�T�4�8�����i�i�k������M�M�J�K�����e�$�����c�"��5�	��M�M�#����	�s�A(�(
B
�2B�B
r�z$test needs threading.get_native_id()c��Sn[SU5up#nURUS5 URUS5 UR5R5nUR[	U5S5 URUSUS5 g)NaBif True:
            import threading
            import os
            from test import support

            parent_thread_native_id = threading.current_thread().native_id
            print(parent_thread_native_id, flush=True)
            assert parent_thread_native_id == threading.get_native_id()
            childpid = os.fork()
            if childpid == 0:
                print(threading.current_thread().native_id, flush=True)
                assert threading.current_thread().native_id == threading.get_native_id()
            else:
                try:
                    assert parent_thread_native_id == threading.current_thread().native_id
                    assert parent_thread_native_id == threading.get_native_id()
                finally:
                    support.wait_process(childpid, exitcode=0)
            r@rr!rr;)r
r�rNr�r�r�)r6�scriptrDrErFr�s      r�test_native_id_after_fork�%ThreadTests.test_native_id_after_fork�s{����&(��f�5��������Q������c�"��Y�Y�[�+�+�-�
�����Z��!�,����J�q�M�:�a�=�9r!c��[R"SSS9(aURS5 [R"S5n[SU5 g)NT)r�memoryz\https://github.com/python/cpython/issues/124878 - Known race condition that TSAN identifies.a�
            import os
            import sys
            import threading
            import time
            import _testcapi

            lock = threading.Lock()
            lock.acquire()
            thread_started_event = threading.Event()
            def thread_func():
                try:
                    thread_started_event.set()
                    _testcapi.finalize_thread_hang(lock.acquire)
                finally:
                    # Control must not reach here.
                    os._exit(2)

            t = threading.Thread(target=thread_func)
            t.daemon = True
            t.start()
            thread_started_event.wait()
            # Sleep to ensure daemon thread is blocked on `lock.acquire`
            #
            # Note: This test is designed so that in the unlikely case that
            # `0.1` seconds is not sufficient time for the thread to become
            # blocked on `lock.acquire`, the test will still pass, it just
            # won't be properly testing the thread behavior during
            # finalization.
            time.sleep(0.1)

            def run_during_finalization():
                # Wake up daemon thread
                lock.release()
                # Sleep to give the daemon thread time to crash if it is going
                # to.
                #
                # Note: If due to an exceptionally slow execution this delay is
                # insufficient, the test will still pass but will simply be
                # ineffective as a test.
                time.sleep(0.1)
                # If control reaches here, the test succeeded.
                os._exit(0)

            # Replace sys.stderr.flush as a way to run code during finalization
            orig_flush = sys.stderr.flush
            def do_flush(*args, **kwargs):
                orig_flush(*args, **kwargs)
                if not sys.is_finalizing:
                    return
                sys.stderr.flush = orig_flush
                run_during_finalization()

            sys.stderr.flush = do_flush

            # If the follow exit code is retained, `run_during_finalization`
            # did not run.
            sys.exit(1)
        r@)rrrBrNrOr
�r6r^s  r� test_finalize_daemon_thread_hang�,ThreadTests.test_finalize_daemon_thread_hang�sG���"�"�$�t�<�
�M�M�<�
=����:"�:
��v	��v�&r!rKN)F)=rFrGrHrI�maxDiffrr�r�r�r�r�r�r�r�r�rr.r<rGrJrOr[r�
bigmemtestrmrtr�r�r�r�r r�r�r�r�
skipUnlessr�r�r�r�r�r�r�r�r�r�r	rrr'r0r4rErKrPrTr[r*r_rdrJrKr!rrrrrxs���G��=��=�4�6E�/�F�".�H,�	 �	 �2u�lU�p
F�#!�J�<#�(0�"
���R��%�@� �A� �4
%�N�B&�>)�"��#��#�>�;��;�,��
�����Y�/�1J�K�3�L��3�2�
�����Y�/�1J�K�%�L��%�N�
�����Y�/�1J�K�<7�L��<7�|P�5�:!#�F�66�$�&*��&*�P
*�4�:.�8�:H�"$�&2�#�BO�"!#�H�
�����O�<�>d�e�:�f��:�6�C'��C'r!rrc���\rSrSrSrSr\S5r\S5r\	R"\R\
;S5\R"SSS	S
9S55rSr\S
5r\S5rSrg)�ThreadJoinOnShutdownic��SU-n[SU5up#nUR5RSS5nURUS5 g)Na�if 1:
            import sys, os, time, threading

            # a thread, which waits for the main program to terminate
            def joiningfunc(mainthread):
                mainthread.join()
                print('end of thread')
                # stdout is fully buffered because not a tty, we have to flush
                # before exit.
                sys.stdout.flush()
        
r@r�r�zend of main
end of thread
r�)r6r^rDrErFr�s      r�
_run_and_join�"ThreadJoinOnShutdown._run_and_join!sN��
��
��(��f�5�����z�z�|�#�#�D�"�-������=�>r!c�*�SnURU5 g)Nz�if 1:
            import os
            t = threading.Thread(target=joiningfunc,
                                 args=(threading.current_thread(),))
            t.start()
            time.sleep(0.1)
            print('end of main')
            �rlrcs  r�test_1_join_on_shutdown�,ThreadJoinOnShutdown.test_1_join_on_shutdown2s����	
���6�"r!c�*�SnURU5 g)Na�if 1:
            from test import support

            childpid = os.fork()
            if childpid != 0:
                # parent process
                support.wait_process(childpid, exitcode=0)
                sys.exit(0)

            # child process
            t = threading.Thread(target=joiningfunc,
                                 args=(threading.current_thread(),))
            t.start()
            print('end of main')
            rorcs  r�test_2_join_in_forked_process�2ThreadJoinOnShutdown.test_2_join_in_forked_process>s����	
���6�"r!c�*�SnURU5 g)Na�if 1:
            from test import support

            main_thread = threading.current_thread()
            def worker():
                childpid = os.fork()
                if childpid != 0:
                    # parent process
                    support.wait_process(childpid, exitcode=0)
                    sys.exit(0)

                # child process
                t = threading.Thread(target=joiningfunc,
                                     args=(main_thread,))
                print('end of main')
                t.start()
                t.join() # Should not block: main_thread is already stopped

            w = threading.Thread(target=worker)
            w.start()
            rorcs  r�!test_3_join_in_forked_from_thread�6ThreadJoinOnShutdown.test_3_join_in_forked_from_threadRs��
��*	
���6�"r!zdue to known OS bug�(i`Fr^c��[R"SS9(aURS5 Sn[SU5up4nUR	U5 g)NTrr?a�if True:
            import os
            import random
            import sys
            import time
            import threading

            thread_has_run = set()

            def random_io():
                '''Loop for a while sleeping random tiny amounts and doing some I/O.'''
                import test.test_threading as mod
                while True:
                    with open(mod.__file__, 'rb') as in_f:
                        stuff = in_f.read(200)
                        with open(os.devnull, 'wb') as null_f:
                            null_f.write(stuff)
                            time.sleep(random.random() / 1995)
                    thread_has_run.add(threading.current_thread())

            def main():
                count = 0
                for _ in range(40):
                    new_thread = threading.Thread(target=random_io)
                    new_thread.daemon = True
                    new_thread.start()
                    count += 1
                while len(thread_has_run) < count:
                    time.sleep(0.001)
                # Trigger process shutdown
                sys.exit(0)

            main()
            r@)rrrBr
r�)r6r_r^rDrErFs      r�test_4_daemon_threads�*ThreadJoinOnShutdown.test_4_daemon_threadsnsI���"�"�$�/�
�M�M�9�:�!��D(��f�5��������r!c��Sn[SU5up#nURUS5 URUR5S5 URUS5 g)Na�if True:
            import threading
            import time

            def thread2():
                time.sleep(0.05)
                print("OK")

            def thread1():
                time.sleep(0.05)
                t2 = threading.Thread(target=thread2)
                t2.start()

            t = threading.Thread(target=thread1)
            t.start()
            # do not join() -- the interpreter waits for non-daemon threads to
            # finish.
            r@r!rSrrM�r6r^rDrErFs     r�test_thread_from_thread�,ThreadJoinOnShutdown.test_thread_from_thread�sQ����$(��f�5��������c�"��������e�,�����Q�r!c�4�Sn[R"[SS9 /n[S5H8n[R
"US9nUR
U5 UR5 M: UHnUR5 M SSS5 g!,(df   g=f)Nc��[R"5nUS:�a[R"USS9 g[R"S5 g)Nr�2r�)r�r�rr�r�)r�s r�do_fork_and_wait�IThreadJoinOnShutdown.test_reinit_tls_after_fork.<locals>.do_fork_and_wait�s/���'�'�)�C��Q�w��$�$�S�2�6�����r!r�r��r�)	r�r�r�r�r*rRr�r�r�)r6r�r�r�r�s     r�test_reinit_tls_after_fork�/ThreadJoinOnShutdown.test_reinit_tls_after_fork�s}��
	��
$�
$�.@�,4�6��G��2�Y���$�$�,<�=�����q�!����	��
�������6�6�6�s�A#B	�	
Bc�n�/n[S5H9n[R"SS9nURU5 UR	5 M; [
R"[SS9 [R"5nUS:XaP[[R"55S:Xa[R"S5 O,[R"S	5 O[R"USS
9 SSS5 UHnUR!5 M g!,(df   N(=f!UHnUR!5 M f=f)Nr�c�.�[R"S5$)Ng333333�?rdrKr!rr��KThreadJoinOnShutdown.test_clear_threads_states_after_fork.<locals>.<lambda>�s�����C�r!r�r�r�rr;�3�4r�)r�r*rRr�r�r�r�r�r�r�r�r�_current_framesr�rr�r�)r6r�r�r�r�s     r�$test_clear_threads_states_after_fork�9ThreadJoinOnShutdown.test_clear_threads_states_after_fork�s���
���r��A�� � �(@�A�A��N�N�1��
�G�G�I��
	��(�(�2D�08�:��g�g�i���!�8��3�.�.�0�1�Q�6������������(�(��r�:�:��������:�:���������s%�D�%BD�&D�
D�D�D4rKN)rFrGrHrIrlrpr rsrvrr#rrrrrgrzr~r�r�rJrKr!rrjrjs���?�"
#��#��#�&�#��#�6�_�_�S�\�\�%6�6�8M�N����R��%�@�,�A�O�,�\ �0����2���r!rjc�|�\rSrSrSrSrSr\S5r\	S5r
SSSSS	.S
jjr\	S5r\	S5r
S
rg)�SubinterpThreadingTestsi�c��[R"5upUR[RU5 UR[RU5 [	[S5(a[R
"US5 X4$)N�set_blockingF)r��piper(�closer�r�)r6�r�ws   rr��SubinterpThreadingTests.pipe�sU���w�w�y���������!�$�������!�$��2�~�&�&��O�O�A�u�%��v�
r!c��UR5up[R"SU4-5n[RRU5nUR
US5 UR
[R"US5S5 g)Na�
            import os
            import random
            import threading
            import time

            def random_sleep():
                seconds = random.random() * 0.010
                time.sleep(seconds)

            def f():
                # Sleep a bit so that the thread is still running when
                # Py_EndInterpreter is called.
                random_sleep()
                os.write(%d, b"x")

            threading.Thread(target=f).start()
            random_sleep()
        rr;�x�	r�rNrOrr�run_in_subinterpr�r��read�r6r�r�r�r-s     r�test_threads_join�)SubinterpThreadingTests.test_threads_join�sp���y�y�{������ �$�d�% ���&�l�l�+�+�D�1������a� ��������A���-r!c��UR5up[R"SU4-5n[RRU5nUR
US5 UR
[R"US5S5 g)Na�
            import os
            import random
            import threading
            import time

            def random_sleep():
                seconds = random.random() * 0.010
                time.sleep(seconds)

            class Sleeper:
                def __del__(self):
                    random_sleep()

            tls = threading.local()

            def f():
                # Sleep a bit so that the thread is still running when
                # Py_EndInterpreter is called.
                random_sleep()
                tls.x = Sleeper()
                os.write(%d, b"x")

            threading.Thread(target=f).start()
            random_sleep()
        rr;r�r�r�s     r�test_threads_join_2�+SubinterpThreadingTests.test_threads_join_2sp��
�y�y�{������ �2�d�3 ���4�l�l�+�+�D�1������a� ��������A���-r!c��UR5upSnSnSn[R"5nURSUSU<SUSU<SUSU<S3
5 UR	5 UR[R"US	5U5 UR[R"US	5U5 UR[R"US	5U5 g)
N�I�F�Dz�if True:
            import os
            import threading
            import time

            done = False

            def notify_fini():
                global done
                done = True
                os.write(z, z�)
                t.join()
            threading._register_atexit(notify_fini)

            def task():
                while not done:
                    time.sleep(0.1)
                os.write(z\)
            t = threading.Thread(target=task)
            t.start()

            os.write(�)
            r;)r�r�create�execr�r�r�r�)r6�r_interp�w_interp�INTERP�FINI�DONE�interps       r�test_threads_join_with_no_main�6SubinterpThreadingTests.test_threads_join_with_no_main7s���!�Y�Y�[����������$�$�&�����
�#��2�d�X�.�#��2�d�X�.��Z�r�&��,
�+�	�.	�����������1�-�v�6��������1�-�t�4��������1�-�t�4r!c�2�[S5 S[RRS3nSU<S3n[RR	5 [SU5up4nSSS5 UR
SWR55 g!,(df   N/=f)Nr�z�if 1:
            import os
            import threading
            import time

            def f():
                # Make sure the daemon thread is still running when
                # Py_EndInterpreter is called.
                time.sleep(zJ)
            threading.Thread(target=f, daemon=True).start()
            zKif 1:
            import _testcapi

            _testcapi.run_in_subinterp(r�r@z:Fatal Python error: Py_EndInterpreter: not the last thread)r	rrr)�SuppressCrashReportrr�r�)r6�subinterp_coder^rDrErFs      r�test_daemon_threads_fatal_error�7SubinterpThreadingTests.test_daemon_threads_fatal_error]s����k�"��!�L�L�6�6�7�8
�
��"�	$��
�\�\�
-�
-�
/�0��v�>�L�B�S�0��
�
�,�-0�Z�Z�\�	;�0�
/�s�B�
BTF��allowed�daemon_allowedr(c��[S5 [R"SUSUS35n[[R
5n[R"SU<SUSUSUS	3	5n[RR5 [S
U5u p�SSS5 U	R5$!,(df   W	R5$=f)N�_testinternalcapiz�
            import test.support
            import threading
            def func():
                print('this should not have run!')
            t = threading.Thread(target=func, daemon=r�z#
            t.start()
            zh
            import test.support
            test.support.run_in_subinterp_with_config(
                z�,
                use_main_obmalloc=True,
                allow_fork=True,
                allow_exec=True,
                allow_threads=z',
                allow_daemon_threads=z0,
                check_multi_interp_extensions=z;,
                own_gil=False,
            )
            r@)
r	rNrO�boolr�Py_GIL_DISABLEDrr�r
r�)
r6�before_startr�r�r(r��check_multi_interp_extensionsr^r�rFs
          r�_check_allowed�&SubinterpThreadingTests._check_allowedus���
	�)�*�!���.6�
7=�X�>
�
�N�
�
*���)-�W�-D�-D�(E�%����&� �"�#� '�i�(&�&4�%5�6/�/L�.M�N
�"����\�\�
-�
-�
/�(��v�6�I�A�q�0��z�z�|��0�
/��z�z�|��s�B,�,
C	c�J�URSSSS9nURSU5 g)NFr�r)r�r��r6rFs  r�test_threads_not_allowed�0SubinterpThreadingTests.test_threads_not_allowed�s0���!�!�� ��"�
��
	
�
�
�n�c�*r!c�>�URS5 URSSSS9nURSU5 SSS5 URS5 URSSSSS9nURSU5 SSS5 g!,(df   NM=f!,(df   g=f)Nzvia Thread()TFr�rzvia Thread.daemon setterzt.daemon = True)r�r�r�r�s  r�test_daemon_threads_not_allowed�7SubinterpThreadingTests.test_daemon_threads_not_allowed�s���
�\�\�.�
)��%�%��$��&��C�

�M�M�.�#�.�
*��\�\�4�
5��%�%�!��$��	&��C�
�M�M�.�#�.�6�
5�*�
)��6�
5�s�$A=�%B�=
B�
BrKN)r�)rFrGrHrIr�r�r�r%r�rr�r�r�r�rJrKr!rr�r��sw���.�8#.�J�#5��#5�J�;��;�.�#�&*�#��B�+��+��/��/r!r�c�v�\rSrSrSrSrSrSrSr\	"5S5r
SrS	rS
r
SrSrS
rSrSrg)�ThreadingExceptionTestsi�c��[R"5nUR5 UR[UR5 UR5 grC)r*rRr�r�rr��r6rs  r�test_start_thread_again�/ThreadingExceptionTests.test_start_thread_again�s6���!�!�#���������,����5����
r!c�n�[R"5nUR[UR5 grC)r*r�r�rr�)r6r�s  r�test_joining_current_thread�3ThreadingExceptionTests.test_joining_current_thread�s&��"�1�1�3�����,��(;�(;�<r!c�n�[R"5nUR[UR5 grC)r*rRr�rr�r�s  r�test_joining_inactive_thread�4ThreadingExceptionTests.test_joining_inactive_thread�s$���!�!�#�����,����4r!c��[R"5nUR5 UR[[
USS5 UR
5 gr�)r*rRr�r�rr)r�r�s  r�test_daemonize_active_thread�4ThreadingExceptionTests.test_daemonize_active_thread�s8���!�!�#���������,����4�H����
r!c�n�[R"5nUR[UR5 grC)r*r�r�rr�)r6rqs  r�test_releasing_unacquired_lock�6ThreadingExceptionTests.test_releasing_unacquired_lock�s"���~�~������,����5r!c�t�SnSn[R"[RSU/[R[RS9nUR5upEUR
5RSS5nURURSSUR
5-5 URXb5 g)	Naif True:
            import threading

            def recurse():
                return recurse()

            def outer():
                try:
                    recurse()
                except RecursionError:
                    pass

            w = threading.Thread(target=outer)
            w.start()
            w.join()
            print('end of main thread')
            zend of main thread
r@)�stdout�stderrr�r�rzUnexpected error: )
�
subprocess�Popenr�
executable�PIPE�communicater�r�r��
returncode)r6r^�expected_output�pr�r�r�s       r�test_recursion_limit�,ThreadingExceptionTests.test_recursion_limit�s�����"1�����c�n�n�d�F�;�$.�O�O�J�O�O�
M���������}�}��&�&�t�R�0��������q�*>�����*P�Q�����/r!c��Sn[SU5up#nURUS5 UR5nURSU5 URSU5 URSU5 UR	SU5 g)Na�if True:
            import threading
            import time

            running = False
            def run():
                global running
                running = True
                while running:
                    time.sleep(0.01)
                1/0
            t = threading.Thread(target=run)
            t.start()
            while not running:
                time.sleep(0.01)
            running = False
            t.join()
            r@r!�Exception in thread�"Traceback (most recent call last):�ZeroDivisionError�Unhandled exception�r
r�r�r�r�r}s     r�test_print_exception�,ThreadingExceptionTests.test_print_exception�st����$(��f�5��������c�"��j�j�l���
�
�+�S�1��
�
�:�C�@��
�
�)�3�/����.��4r!c��Sn[SU5up#nURUS5 UR5nURSU5 URSU5 URSU5 UR	SU5 g)Na�if True:
            import sys
            import threading
            import time

            running = False
            def run():
                global running
                running = True
                while running:
                    time.sleep(0.01)
                1/0
            t = threading.Thread(target=run)
            t.start()
            while not running:
                time.sleep(0.01)
            sys.stderr = None
            running = False
            t.join()
            r@r!r�r�r�r�r�r}s     r�%test_print_exception_stderr_is_none_1�=ThreadingExceptionTests.test_print_exception_stderr_is_none_1	st����((��f�5��������c�"��j�j�l���
�
�+�S�1��
�
�:�C�@��
�
�)�3�/����.��4r!c��Sn[SU5up#nURUS5 URSUR55 g)Na�if True:
            import sys
            import threading
            import time

            running = False
            def run():
                global running
                running = True
                while running:
                    time.sleep(0.01)
                1/0
            sys.stderr = None
            t = threading.Thread(target=run)
            t.start()
            while not running:
                time.sleep(0.01)
            running = False
            t.join()
            r@r!r�)r
r�r�r�r}s     r�%test_print_exception_stderr_is_none_2�=ThreadingExceptionTests.test_print_exception_stderr_is_none_2&sB����((��f�5��������c�"����.��
�
��=r!c� �Sn[SU5 g)Na9if True:
            import time
            import threading
            import _thread

            def f():
                try:
                    f()
                except RecursionError:
                    f()

            def g():
                try:
                    raise ValueError()
                except* ValueError:
                    f()

            def h():
                time.sleep(1)
                _thread.interrupt_main()

            t = threading.Thread(target=h)
            t.start()
            g()
            t.join()
            r@)rrcs  r�test_print_exception_gh_102056�6ThreadingExceptionTests.test_print_exception_gh_102056?s����6	�d�F�+r!c�^�Sm"U4SjS[R5nU"5nUR5 UR5 UR	UR
5 UR
UR
[5 SUlg)Nc��erCrKrKr!r�
bare_raise�OThreadingExceptionTests.test_bare_raise_in_brand_new_thread.<locals>.bare_raise_s��r!c�&>�\rSrSrSrU4SjrSrg)�OThreadingExceptionTests.test_bare_raise_in_brand_new_thread.<locals>.Issue27558ibNc�P>�T"5 g![anXlSnAgSnAff=frC)r$�exc)r6rr�s  �rrb�SThreadingExceptionTests.test_bare_raise_in_brand_new_thread.<locals>.Issue27558.runes$���#��L�� �#�"�H�H��#�s��
%� �%)r)rFrGrHrIrrbrJ)r�s�r�
Issue27558r�bs����C�
#�
#r!r)r*rRr�r�r�rr�r)r6rrr�s   @r�#test_bare_raise_in_brand_new_thread�;ThreadingExceptionTests.test_bare_raise_in_brand_new_thread^sb���	�	#��)�)�	#�����������
����V�Z�Z�(����f�j�j�,�7���
r!c��SnUR[R[R5 [	S5Vs/sHn[
R"US9PM nnUH#nUR5 UR5 M% gs snf)Nc��[[RSSS9nURS5 [R
"5 SSS5 g!,(df   g=f)Nr�r�)�encoding� )�openr�TESTFN�write�	traceback�format_stack)�fps r�modify_file�QThreadingExceptionTests.test_multithread_modify_file_noerror.<locals>.modify_fileus9���i�&�&��g�>�"�����
��&�&�(�?�>�>�s�'A
�

ArRr�)	r(r�unlinkrr�r*rRr�r�)r6rr�r�r�s     r�$test_multithread_modify_file_noerror�<ThreadingExceptionTests.test_multithread_modify_file_noerrorssw��	)�
	
���	�(�(�)�*:�*:�;��3�Z�
���
���K�0��	�
��A�
�G�G�I�
�F�F�H���	
s�B	c��Sn[SU5up#nURUS5 URUS5 URUS5 g)Nz�if True:
        import sys
        import threading

        threading.x = sys.modules[__name__]
        x = threading._DummyThread()
        r@rr!r�r�s     r�)test_dummy_thread_on_interpreter_shutdown�AThreadingExceptionTests.test_dummy_thread_on_interpreter_shutdown�sK��
��(��d�3��������Q������c�"�����c�"r!rKN)rFrGrHrIr�r�r�r�r�rr�r�r�r�r�rrrrJrKr!rr�r��sU���>�5��6���0��0�>5�65�:>�2,�>�*� #r!r�c��\rSrSrSrSrg)�
ThreadRunFaili�c��[S5e)N�
run failed�r�r5s rrb�ThreadRunFail.run�s
����&�&r!rKNrrKr!rrr�s��'r!rc�~^�\rSrSrU4Sjr\S5r\R\S55r	Sr
SrSrSr
S	rU=r$)
�ExceptHookTestsi�c�8>�[U5 [TU]	5 grC)r-�superrj)r6�	__class__s �rrj�ExceptHookTests.setUp�s���"�4�(�
��
�r!c��[R"S5n[SS9nUR5 UR	5 SSS5 WR5R
5nURSWRS3U5 URSU5 URSU5 URSU5 g!,(df   N�=f)	Nr��excepthook threadrP�Exception in thread �:
�#Traceback (most recent call last):
z   raise ValueError("run failed")zValueError: run failed)	r�captured_outputrr�r��getvaluerNr�rQ)r6r�rs   r�test_excepthook�ExceptHookTests.test_excepthook�s���
�
$�
$�X�
.�&�"�(;�<�F��L�L�N��K�K�M�/�
���"�(�(�*���
�
�,�V�[�[�M��=�v�F��
�
�<�f�E��
�
�8�&�A��
�
�.��7�/�
.�s�*B>�>
Cc�B�[R"S5n[S5e![aWn[R
"/[R"5QSP5n[R"U5 SnSnAO!Snf=fSnAff=fSSS5 O!,(df   O=fWR5R5nURS[R"5S3U5 URSU5 URSU5 URSU5 g)Nr��bugr&r'r(z  raise ValueError("bug")zValueError: bug)
rr)r�r$r*�ExceptHookArgsr�exc_infor'r*rNr�r)r6r�rr�s    r�test_excepthook_thread_None�+ExceptHookTests.test_excepthook_thread_None�s���
�
$�
$�X�
.�&�
 � ��'�'���
 � �/�/�0G�#�,�,�.�0G�$�0G�H�� ��(�(��.� �D��4�D��

 ���/�
.�
.�����"�(�(�*���
�
�,�Y�-@�-@�-B�,C�3�G��P��
�
�<�f�E��
�
�1�6�:��
�
�'��0sC�B�$�
B�-B�A9�2B�4B�9A=�=B�B�B�
B c�$�"SS[R5n[R"S5nU"5nUR	5 UR5 SSS5 UR
WR5S5 g!,(df   N/=f)Nc��\rSrSrSrSrg)�4ExceptHookTests.test_system_exit.<locals>.ThreadExiti�c�0�[R"S5 gr:)r�exitr5s rrb�8ExceptHookTests.test_system_exit.<locals>.ThreadExit.run�s
������r!rKNrrKr!r�
ThreadExitr5�s��
r!r9r�r�)r*rRrr)r�r�r�r*)r6r9r�rs    r�test_system_exit� ExceptHookTests.test_system_exit�sf��	��)�)�	�
�
$�
$�X�
.�&��\�F��L�L�N��K�K�M�/�
	
������*�B�/�/�
.�s�(B�
Bc�^�SmU4Sjn[R"[SU5 [5nUR	5 UR5 SSS5 UR
TR[5 UR
[TR5S5 UR
TRTRR5 URTRW5 Smg!,(df   N�=f!Smf=f)Nc�
>�UmgrCrK)�	hook_argsr�s �r�hook�4ExceptHookTests.test_custom_excepthook.<locals>.hook�s����Dr!r'r)r�	swap_attrr*rr�r�r��exc_typer�r��	exc_value�
exc_traceback�
__traceback__r�r)r6r?rr�s   @r�test_custom_excepthook�&ExceptHookTests.test_custom_excepthook�s������	�	��"�"�9�l�D�A�&����������
�B�

���T�]�]�J�7����S����0�,�?����T�/�/����1M�1M�N��M�M�$�+�+�v�.��D�B�A���D�s#�C?�+C.�BC?�.
C<�8C?�?Dc�^�SnSmU4Sjn[R"[SU5 [R"[SU5 [R"S5n[5nUR
5 UR5 SSS5 SSS5 SSS5 URWR5S5 URTS5 g!,(df   NQ=f!,(df   NZ=f!,(df   Nc=f)Nc��[S5e)N�threading_hook failedrrs r�threading_hook�CExceptHookTests.test_custom_excepthook_fail.<locals>.threading_hook�s���4�5�5r!c�>�[U5mgrCr�)rBrCrD�err_strs   �r�sys_hook�=ExceptHookTests.test_custom_excepthook_fail.<locals>.sys_hook�s����)�n�Gr!r'r�z#Exception in threading.excepthook:
rJ)
rrAr*rr)rr�r�r�r*)r6rKrOr�rrNs     @r�test_custom_excepthook_fail�+ExceptHookTests.test_custom_excepthook_fail�s����	6���	%��
�
�y�,��
G�
�
�
�s�L�(�
;�
�
$�
$�X�
.�&�"�_�F��L�L�N��K�K�M�/�<�H�	
������*�?�	A�����"9�:�/�
.��<�
;��H�
G�s;�C5�C$�+C�C$�C5�
C!�C$�$
C2	�.C5�5
Dc�D�SnSnU"5n[R"[SU5 U"5n[R[lU"5nSSS5 URUW5 UR
UW5 URUS5 g!,(df   NE=f)Nc���[R"S5n[SS9nUR5 UR	5 SSS5 UR5$!,(df   WR5$=f)Nr�r%rP)rr)rr�r�r*)�outputrs  r�
run_thread�<ExceptHookTests.test_original_excepthook.<locals>.run_thread�s[���(�(��2�f�&�,?�@���������
�3��?�?�$�$�	3�2��?�?�$�$�s�*A�
A6c�4�[S[RS9 g)NzRunning a thread failed)�file)r\rr�rs rrK�@ExceptHookTests.test_original_excepthook.<locals>.threading_hooks���+�#�*�*�=r!r'zRunning a thread failed
)rrAr*r+r'r�r�)r6rVrK�default_output�custom_hook_output�recovered_outputs      r�test_original_excepthook�(ExceptHookTests.test_original_excepthook�s���	%�	>�$���
�
�
�y�,��
G�!+���#,�#;�#;�I� �)�|��H�
	
����)9�:����N�,>�?����+�-H�I�H�
G�s�(B�
BrK)rFrGrHrIrjrr+rrr1r:rFrQr^rJ�
__classcell__�r"s@rrr�sY�����
8��
8�
����1���1�(0��*;�*J�Jr!rc�&�\rSrSrSrSrSrSrg)�
TimerTestsic�p�[RU5 /Ul[R"5UlgrC)rerj�
callback_argsr*r��callback_eventr5s rrj�TimerTests.setUps'�����4� ����'�o�o�/��r!c��[R"SUR5nUR5 URR5 URRS5 SURS'URR5 [R"SUR5nUR5 URR5 UR[UR5S5 URURS04S04/5 UR5 UR5 g)Nrk�blah�bar�foorrK)r*�Timer�
_callback_spyr�rfr�r�r�r4rr�r�rer�)r6�timer1�timer2s   r� test_init_immutable_default_args�+TimerTests.test_init_immutable_default_argss�������t�'9�'9�:��������� � �"������6�"�$��
�
�e�����!�!�#�����t�'9�'9�:��������� � �"�����T�/�/�0�!�4�����+�+�r�2�h��R��-A�B����
����
r!c��URRUSSUR545 URR	5 grC)rer��copyrfr�)r6r�r4s   rrm�TimerTests._callback_spy's7�����!�!�4��7�F�K�K�M�":�;������!r!)rerfN)rFrGrHrIrjrprmrJrKr!rrcrcs��0�
�""r!rcc�8�\rSrSr\"\R5rSrg)�	LockTestsi+rKN)	rFrGrHrI�staticmethodr*r��locktyperJrKr!rrvrv+s���I�N�N�+�Hr!rvc�8�\rSrSr\"\R5rSrg)�PyRLockTestsi.rKN)	rFrGrHrIrwr*�_PyRLockrxrJrKr!rrzrz.s���I�.�.�/�Hr!rzzRLock not implemented in Cc�>�\rSrSr\"\R5rSrSr	g)�CRLockTestsi1c�x�[R"SS9n[R"5 SSS5 UR	W/5 S04SSS04SSS04/nUHQup4URX4S9 UR
[5 [R"U0UD6 SSS5 SSS5 MS "S	S
UR5n[R"SS9nU"SSS9 SSS5 UR	U/5 g!,(df   N�=f!,(df   Nw=f!,(df   M�=f!,(df   NU=f)
NT)�recordr�rKr�r;)r;rr3c�,^�\rSrSrSU4SjjrSrU=r$)�/CRLockTests.test_signature.<locals>.CustomRLockiEc�">�[TU]5 grC)r!r7)r6r�r�r"s   �rr7�8CRLockTests.test_signature.<locals>.CustomRLock.__init__Fs����� �"r!rK)�returnN)rFrGrHrIr7rJr`ras@r�CustomRLockr�Es
���
#�
#r!r�r)r�)	r�r�r*r�r�r��assertWarnsr�rx)r6�warnings_log�	arg_typesr�r4r�s      r�test_signature�CRLockTests.test_signature5s��
�
$�
$�D�
1�\��O�O��2�����r�*��2�J�
�#�q��N�
�c�1�X��
�	�
&�L�D����4��7��%�%�&8�9��O�O�T�4�V�4�:�8�7�&�	#�$�-�-�	#��
$�
$�D�
1�\���Q��2�����r�*�+2�
1��:�9��8�7��2�
1�s;�C7�*D�D�D�D+�7
D�
D�D�
D(	�+
D9rKN)
rFrGrHrIrwr*�_CRLockrxr�rJrKr!rr}r}1s���I�-�-�.�H�+r!r}c�8�\rSrSr\"\R5rSrg)�
EventTestsiMrKN)	rFrGrHrIrwr*r��	eventtyperJrKr!rr�r�Ms���Y�_�_�-�Ir!r�c�>�\rSrSr\"\R5rSrSr	g)�ConditionAsRLockTestsiPc�&�URS5 g)Nz,Condition does not expose _recursion_count())rBr5s r�test_recursion_count�*ConditionAsRLockTests.test_recursion_countTs���
�
�D�Er!rKN)
rFrGrHrIrwr*r�rxr�rJrKr!rr�r�Ps���I�/�/�0�H�Fr!r�c�8�\rSrSr\"\R5rSrg)�ConditionTestsiWrKN)	rFrGrHrIrwr*r��condtyperJrKr!rr�r�Ws���I�/�/�0�Hr!r�c�8�\rSrSr\"\R5rSrg)�SemaphoreTestsiZrKN)	rFrGrHrIrwr*�	Semaphore�semtyperJrKr!rr�r�Zs���9�.�.�/�Gr!r�c�8�\rSrSr\"\R5rSrg)�BoundedSemaphoreTestsi]rKN)	rFrGrHrIrwr*r�r�rJrKr!rr�r�]s���9�5�5�6�Gr!r�c�8�\rSrSr\"\R5rSrg)�BarrierTestsi`rKN)	rFrGrHrIrwr*�Barrier�barriertyperJrKr!rr�r�`s���y�0�0�1�Kr!r�c��\rSrSrSrSrg)�MiscTestCaseidc�^�[U5 S1nSS1n[R"U[SXS9 g)Nr2r�r�)r*r�)�extra�not_exported)r-r�check__all__r*)r6r�r�s   r�test__all__�MiscTestCase.test__all__es4��"�4�(����'��7�����T�9�.F�#(�	Er!rKN)rFrGrHrIr�rJrKr!rr�r�ds��Er!r�c�x�\rSrSrSrSr\"S5S5rSrSr	Sr
S	r\RS
5rSrg)
�InterruptMainTestsinc��Sn[R"X5nUR[RX5 UR[5 [R
"5 SSS5 g!,(df   g=f)Nc��SS- g)Nr;rrK)�signumr�s  r�handler�LInterruptMainTests.check_interrupt_main_with_signal_handler.<locals>.handlerps	��
�a�Cr!)�signalr(r�r�r��interrupt_main)r6r�r��old_handlers    r�(check_interrupt_main_with_signal_handler�;InterruptMainTests.check_interrupt_main_with_signal_handlerosP��	��m�m�F�4������
�
�v�;�
�
�
�0�
1��"�"�$�2�
1�
1�s�A.�.
A<c��[R"U5n[R"U[R5 [R"U5 [R"U[R
5 [R"U5 [R"X5 g![R"X5 f=frC)r��	getsignal�SIG_IGNr�r��SIG_DFL)r6r�r�s   r�check_interrupt_main_noerror�/InterruptMainTests.check_interrupt_main_noerrorysr���"�"�6�*��		+��M�M�&�&�.�.�1��"�"�6�*��M�M�&�&�.�.�1��"�"�6�*�
�M�M�&�*��F�M�M�&�*�s�A6B%�%B=z*gh-118433: Flaky due to a longstanding bugc���Sn[R"US9nUR[5 UR	5 UR5 SSS5 UR5 g!,(df   N=f)Nc�.�[R"5 grC)r�r�rKr!r�call_interrupt�HInterruptMainTests.test_interrupt_main_subthread.<locals>.call_interrupt�s���"�"�$r!r�)r*rRr��KeyboardInterruptr�r�)r6r�r�s   r�test_interrupt_main_subthread�0InterruptMainTests.test_interrupt_main_subthread�sQ��	%����N�3��
�
�
�0�
1�
�G�G�I�
�F�F�H�2�	
����2�
1�s�!A'�'
A5c��UR[5 [R"5 SSS5 g!,(df   g=frC)r�r�r�r�r5s r�test_interrupt_main_mainthread�1InterruptMainTests.test_interrupt_main_mainthread�s+���
�
�0�
1��"�"�$�2�
1�
1�s	�5�
Ac��UR[R5 UR[R5 grC)r�r��SIGINT�SIGTERMr5s r�'test_interrupt_main_with_signal_handler�:InterruptMainTests.test_interrupt_main_with_signal_handler�s&���5�5�f�m�m�D��5�5�f�n�n�Er!c��UR[R5 UR[R5 grC)r�r�r�r�r5s r�test_interrupt_main_noerror�.InterruptMainTests.test_interrupt_main_noerror�s&���)�)�&�-�-�8��)�)�&�.�.�9r!c��UR[[RS5 UR[[R[R
5 UR[[RS5 g)Nri@B)r�r�r�r�r��NSIGr5s r�"test_interrupt_main_invalid_signal�5InterruptMainTests.test_interrupt_main_invalid_signal�sN�����*�g�&<�&<�b�A����*�g�&<�&<�f�k�k�J����*�g�&<�&<�g�Fr!c���S/nS/nS/nSn[R"XBX4S9nUR5 US(d
US(dM
SUS'UR5 UR	US5 g)NTFc�f�SnSUS'US(aU(aUS-nOgUS(aMSUS'g)Ni�Trr;rK)r��cont�interrupted�
iterationss    rre�AInterruptMainTests.test_can_interrupt_tight_loops.<locals>.worker�s?��$�J��G�A�J��q�'���!�O�J����q�'�'�"�K��Nr!r�r)r*rRr�r�r�)r6r�r�r�rer�s      r�test_can_interrupt_tight_loops�1InterruptMainTests.test_can_interrupt_tight_loops�ss���v���'���g��		"�
���F�$�0L�M��	���	��!�*���!�*�*���Q��	��������A��'r!rKN)rFrGrHrIr�r�rr�r�r�r�r�r�reap_threadsr�rJrKr!rr�r�nsW��%�+��F�G�	�H�	�%�F�:�G�
�"�"�(�#�(r!r�c�&�\rSrSrSrSrSrSrg)�AtexitTestsi�c��[SS5upnURU5 URUR5S5 g)Nr@z�if True:
            import threading

            def run_last():
                print('parrot')

            threading._register_atexit(run_last)
        sparrot)r
r�r�rNrCs    r�test_atexit_output�AtexitTests.test_atexit_output�s=��'��/�
����	
������������i�0r!c�D�[SS5upnURU5 g)Nr@aNif True:
            import threading
            from unittest.mock import Mock

            mock = Mock()
            threading._register_atexit(mock)
            mock.assert_not_called()
            # force early shutdown to ensure it was called once
            threading._shutdown()
            mock.assert_called_once()
        )r
r�rCs    r�test_atexit_called_once�#AtexitTests.test_atexit_called_once�s'��'��
/�

����	
����r!c��[SS5upnURU5 URSUR55 g)Nr@z�if True:
            import threading

            def func():
                pass

            def run_last():
                threading._register_atexit(func)

            threading._register_atexit(run_last)
        z2RuntimeError: can't register atexit after shutdown)r
r�r�r�rCs    r�test_atexit_after_shutdown�&AtexitTests.test_atexit_after_shutdown�s?��(��
/�

����	
������
�
�J��
�
��	r!rKN)rFrGrHrIr�r�r�rJrKr!rr�r��s��1�
�r!r��__main__)D�__doc__�test.supportrrrrrrr�test.support.import_helperr	�test.support.script_helperr
rrr[rr�r*r^rr�r�r�r�rNr
r�r
rrrrZ�requires_working_threadingrr r%r-r�r/rRrM�TestCasererrrjr�r�rrrcrv�
RLockTestsrzr#r�r}r�r�r�r�r�r�r�r�r�rFr�rKr!r�<module>r�sV����T�T�9�9�4�N�,�
�
������	��
��������)��+�+�4�8�+��	�=�4��f��<��!�!�<�B%�8�$�$�%�d'�,�d'�N%K�<�K�\C/�l�C/�L_#�l�_#�D'�I�$�$�'�
rJ�l�rJ�j"��"�8,�
�$�$�,�0�:�(�(�0�
����"�"�d�*�,H�I�+�*�'�'�+�J�+�6.��&�&�.�F�J�1�1�F�1�Z�.�.�1�0�Z�.�.�0�7�J�<�<�7�2�:�*�*�2�E�8�$�$�E�N(��*�*�N(�b/�(�#�#�/�d�z���M�M�O���cF���L��s�6I�I&�%I&

Youez - 2016 - github.com/yon3zu
LinuXploit